package com.yin.myproject.zk.selector;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;

import java.io.Closeable;
import java.io.IOException;

/**
 * Created by Eason on 2017/8/2.
 */
public class SchedulerSelector extends LeaderSelectorListenerAdapter implements Closeable {

    private static final String PATH = "/tasks/leader";

    private final String name;
    private final LeaderSelector leaderSelector;

    public SchedulerSelector(CuratorFramework client, String path, String name) {
        this.name = name;
        leaderSelector = new LeaderSelector(client, path, this);
        leaderSelector.autoRequeue();
    }

    public static void main(String[] args) throws Exception {
        String ips = "10.0.20.183:8080,10.0.20.183:8081,10.0.20.183:8082";
        String[] ipArray = ips.split(",");
        TestingServer server = new TestingServer();
        for (int i = 0; i < ipArray.length; i++) {
            CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
            client.start();
            SchedulerSelector selector = new SchedulerSelector(client, PATH, "Client #" + i);
            selector.start();
        }
    }

    public void start() throws IOException {
        leaderSelector.start();
    }

    @Override
    public void close() throws IOException {

    }

    @Override
    public void takeLeadership(CuratorFramework client) throws Exception {
        System.out.println("*************************************************************************");
        System.out.println(name + "is leader");
        client.create().forPath(PATH + name);
        leaderSelector.close();
    }
}